package com.flowcloud.kafka.config;

import com.flowcloud.common.utils.SpringContextHolder;
import com.flowcloud.kafka.exception.UipListenableFutureCallback;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @Author: wuwenbin
 * @Date：2020/12/18 08:39
 * <p>
 * Description:
 */
public class KafKaSender {
	private final static KafkaTemplate<String, Object> kafkaTemplate = SpringContextHolder.getBean(KafkaTemplate.class);

	public static void sendMessage(String topic, Object msg) {
		sendMessage(topic, null, msg);
	}

	public static void sendMessage(String topic, String key, Object msg) {
		ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(topic, key, msg);
		listenableFuture.addCallback(new UipListenableFutureCallback(topic, System.nanoTime(), msg));
	}
}
